package main

import (
	"fmt"
	"os"
	"time"
	"strconv"
	"strings"
)

// Data structures required for the RPC requests from another node to send the hashtables of the logs from
// this node to that node
type LogArgs struct {
     ip_addr string
     partitionNumber int
     start_point int
}

type LogReply struct {
}

type HashtableArgs struct {
     ip_addr string
     partitionNumber int
}

type HashtableReply struct {
}

type RPCLogRequest int;

// End of data structures for the RPC requests

type log_value struct {
	partitionNumber int
	wlog []log_data
}

var chan_table map[string]chan log_value

func setupExporter(ip_addr string) (*Exporter, os.Error) {
	exp, err := NewExporter("tcp", ip_addr)
	if err != nil {
		fmt.Println("Error in creating new Log Sending Exporter:", err)	
	}
	return exp,err	
}

func setupLogChannel(exp *Exporter) (chan log_value, os.Error) {
	c := make(chan log_value)
	err := exp.Export("log_channel", c, Send)
	if err != nil {
		fmt.Println("Error in setting up log channel:", err)
		return nil, err
	}
	return c,err
}

func setupExporters() bool {
	chan_table = make(map[string]chan log_value)
	for  k:=0; k< n_servers-1; k++ {
		tPort := getTransferPort(peer_servers[k])
		channel, err := createChannel(":"+tPort)
		if err!=nil {
			fmt.Println("Error in setting up exporters:",err.String())
			return false
		}
		chan_table[peer_servers[k]] = channel
	}
	return true		
}

// ip addr has to be of the form "host:port"
func importLogReceive(ip_addr string) os.Error {
	var err2 os.Error
	var ch chan log_value
	for ; ; {
		imp, err := NewImporter("tcp", ip_addr)
		if err != nil {
			//fmt.Println("Error in setting up log receiver:", err)
			continue
		}

		ch = make(chan log_value)
		err2 = imp.ImportNValues("log_channel", ch, Recv, new(log_value), 0)
		if err2 != nil {
			//fmt.Println("Error in setting up log receiver:", err2)	
			continue
		}
		if err == nil && err2==nil {
			break
		}
		time.Sleep(2e9)
	}
	//fmt.Printf("Finished first for loop. Waiting for logs now.\n");
	for ; ; {
		v := <-ch
		//fmt.Println("Got a log value!");
		//fmt.Printf("%v\n", v)
		processLog(v.partitionNumber, v.wlog)
		time.Sleep(1e9)
	}
	return err2
}

func setupImporters() bool {

	//Waiting for exporters to finish exporting
	//time.Sleep(10000000000)
	
	tPort := getTransferPort(my_ip)
	for k := 0; k< n_servers-1; k++ {
		go importLogReceive(peer_servers[k]+":"+tPort)
	}
	return true		
}

func makeLog(hash_table map[string]hash_record) []log_data {
	newlog := make([]log_data,len(hash_table),len(hash_table)+100)
	wp := 0

	for key,hvalue := range hash_table {
		newlog[wp].key= key
		newlog[wp].value = hvalue.value
		newlog[wp].time_written = hvalue.time_written 
		newlog[wp].id = int64(wp)
		wp = wp + 1
	}
	//fmt.Printf("New log: %v\n", newlog);
	return newlog
}

func processLog(partitionNumber int, wlog []log_data) {
	for _, c := range wlog {
		writePartitionWithTime(partitionNumber,c.key,c.value,c.time_written)
	}
}

func sendHashTable_Chan(log_channel chan log_value, partitionNumber int, table map[string]hash_record) {
	var logval log_value
	logval.partitionNumber = partitionNumber
	logval.wlog = makeLog(table)
	time.Sleep(5e9)
	log_channel <- logval
	//fmt.Println("Put the hashtable into the drain, part#", partitionNumber)
}

func sendLog_Chan(log_channel chan log_value, partitionNumber int, wlog []log_data) {
	var logval log_value
	logval.partitionNumber = partitionNumber
	logval.wlog = wlog
	log_channel <- logval
}

// Wrapper to call sendHashTable remotely from another machine
func (t* RPCLogRequest) RPCSendHashTable(args *HashtableArgs, reply *HashtableArgs) os.Error {
     sendHashTable(args.ip_addr, args.partitionNumber);
     return nil;
}

func sendHashTable(ip_addr string, partitionNumber int) {
        //fmt.Printf("In sendhashtable : ip %s partition %d\n", ip_addr, partitionNumber);
	log_channel := chan_table[ip_addr]
	sendHashTable_Chan(log_channel,partitionNumber,hashtable[partitionNumber])
}

func sendLog_Array(ip_addr string, partitionNumber int, wlog []log_data) {
	//fmt.Println("Inside sendLog:",ip_addr,partitionNumber);
	log_channel := chan_table[ip_addr]
	sendLog_Chan(log_channel, partitionNumber, wlog)
}

// Wrapper for RPC request from another node to this one to send its logs to that node.
func (t *RPCLogRequest) RPCSendLog(args *LogArgs, reply *LogReply) os.Error {
     // On receiving an RPC request, send the logs and update entry that he has seen these logs.
     logs_seen_mutex.Lock();     
     start_point := logs_seen_index[args.partitionNumber][ip_serverid_map[args.ip_addr]];
     go sendLog(args.ip_addr, args.partitionNumber, start_point);
     logs_seen_index[args.partitionNumber][ip_serverid_map[args.ip_addr]] = getLogEnd(args.partitionNumber);
     logs_seen_mutex.Unlock();     
     return nil;
}

func sendLog(ip_addr string, partitionNumber int, start_point int) {
	//fmt.Println("sendLog:",start_point,wlogs[partitionNumber].write_pointer)
	if wlogs[partitionNumber].write_pointer == 0 {
		//fmt.Println("No data in partition",partitionNumber,"to write!")
		return
	} else {
		if( start_point < wlogs[partitionNumber].write_pointer ) {

		sendLog_Array(ip_addr,partitionNumber,wlogs[partitionNumber].log[start_point:wlogs[partitionNumber].write_pointer])
		
		}
	}
}

func createChannel(ip_addr string) (chan log_value,os.Error) {
	exp,err := setupExporter(ip_addr)
	if err != nil {
		return nil, err
	}
	log_channel,err2 := setupLogChannel(exp)
	if err2 != nil {
		return nil, err
	}
	fmt.Println(log_channel)
	return log_channel,err2
}

func startRecievers(ip_addr string) {
	go importLogReceive(ip_addr)
}

func getTransferPort(ip_addr string) string {
	subs := strings.Split(ip_addr,".",0)
	int_port,err := strconv.Atoi(subs[len(subs)-1])
	if err!=nil {
		fmt.Println("Error in parsing port:", err.String())
		return "ERROR"
	}
	int_port = int_port + 12000
	return strconv.Itoa(int_port)
}

// Last log is this-1
func getLogEnd(partitionNumber int) int {
	return wlogs[partitionNumber].write_pointer
}
